package com.chenwc.rocketmq.example;

import com.chenwc.rocketmq.prop.RocketMqConnectionProperties;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * 生产者单例模式
 *
 * @author chenwc
 */
public class ProducerSingleton {
    private static final RocketMqConnectionProperties CONNECTION_PROPERTIES = new RocketMqConnectionProperties();
    private static volatile DefaultMQProducer PRODUCER;
    private static volatile TransactionMQProducer TRANSACTION_PRODUCER;

    private ProducerSingleton() {
    }

    /**
     * 创建生产者
     *
     * @param topics 主题列表，使生产者可以在消息发布之前预取主题路由
     * @return DefaultMQProducer
     * @throws MQClientException mqClientException
     */
    private static DefaultMQProducer buildProducer(List<String> topics) throws MQClientException {

        // 配置 AccessKey 和 SecretKey
        SessionCredentials sessionCredentials = new SessionCredentials(
                CONNECTION_PROPERTIES.getAccessKey(),
                CONNECTION_PROPERTIES.getSecretKey());
        // 创建 ACL 认证
        RPCHook aclHook = new AclClientRPCHook(sessionCredentials);
        // 创建 Producer
        DefaultMQProducer producer;
        if (null != topics && !topics.isEmpty()) {
            producer = new DefaultMQProducer("ProducerGroup", aclHook, topics);
        } else {
            producer = new DefaultMQProducer("ProducerGroup", aclHook);
        }
        // 设置 NameSrv 地址
        producer.setNamesrvAddr(CONNECTION_PROPERTIES.getNamesrvAddr());
        // 设置发送超时时间
        producer.setSendMsgTimeout(3000);
        producer.start();
        return producer;
    }

    /**
     * 创建事务生产者
     *
     * @param topics 主题列表，使生产者可以在消息发布之前预取主题路由
     * @return TransactionMQProducer
     * @throws MQClientException mqClientException
     */
    private static TransactionMQProducer buildTransactionProducer(List<String> topics, TransactionListener transactionListener) throws MQClientException {

        // 配置 AccessKey 和 SecretKey
        SessionCredentials sessionCredentials = new SessionCredentials(
                CONNECTION_PROPERTIES.getAccessKey(),
                CONNECTION_PROPERTIES.getSecretKey());
        // 创建 ACL 认证
        RPCHook aclHook = new AclClientRPCHook(sessionCredentials);
        // 创建 Producer
        TransactionMQProducer producer;
        if (null != topics && !topics.isEmpty()) {
            producer = new TransactionMQProducer("ProducerGroup", aclHook, topics);
        } else {
            producer = new TransactionMQProducer("ProducerGroup", aclHook);
        }
        // 设置 NameSrv 地址
        producer.setNamesrvAddr(CONNECTION_PROPERTIES.getNamesrvAddr());
        // 设置发送超时时间
        producer.setSendMsgTimeout(3000);
        if (null != transactionListener) {
            producer.setTransactionListener(transactionListener);
        }
        producer.start();
        return producer;
    }

    /**
     * 获取生产者
     *
     * @param topics 主题列表，使生产者可以在消息发布之前预取主题路由
     * @return DefaultMQProducer
     * @throws MQClientException mqClientException
     */
    public static DefaultMQProducer getInstance(String... topics) throws MQClientException {
        if (null == PRODUCER) {
            synchronized (ProducerSingleton.class) {
                if (null == PRODUCER) {
                    List<String> topicsList = new ArrayList<>(Arrays.asList(topics));
                    PRODUCER = buildProducer(topicsList);
                }
            }
        }
        return PRODUCER;
    }

    /**
     * 获取事务生产者
     *
     * @param topics 主题列表，使生产者可以在消息发布之前预取主题路由
     * @return DefaultMQProducer
     * @throws MQClientException mqClientException
     */
    public static TransactionMQProducer getTransactionInstance(TransactionListener transactionListener, String... topics) throws MQClientException {
        if (null == TRANSACTION_PRODUCER) {
            synchronized (ProducerSingleton.class) {
                if (null == TRANSACTION_PRODUCER) {
                    List<String> topicsList = new ArrayList<>(Arrays.asList(topics));
                    TRANSACTION_PRODUCER = buildTransactionProducer(topicsList, transactionListener);
                }
            }
        }
        return TRANSACTION_PRODUCER;
    }

    /**
     * 关闭生产者
     */
    public static void closeTransactionInstance() {
        if (null != TRANSACTION_PRODUCER) {
            synchronized (ProducerSingleton.class) {
                if (null != TRANSACTION_PRODUCER) {
                    TRANSACTION_PRODUCER.shutdown();
                    TRANSACTION_PRODUCER = null;
                }
            }
        }
    }

    /**
     * 关闭生产者
     */
    public static void closeInstance() {
        if (null != PRODUCER) {
            synchronized (ProducerSingleton.class) {
                if (null != PRODUCER) {
                    PRODUCER.shutdown();
                    PRODUCER = null;
                }
            }
        }
    }
}
